package com.deep.flink.demo.job._3_function;

import lombok.SneakyThrows;
import org.apache.calcite.rel.core.Join;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import study.bigdata.flink.cdc.function.JoinFunction;
import study.bigdata.flink.cdc.function.TestFunction;

public class _2_TableFunctionJob {

    @SneakyThrows
    public static void main(String[] args) {
        //
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .useBlinkPlanner()
                .inStreamingMode()
                .build();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);
        //========== 【注册函数】 ==========
        tableEnv.createTemporarySystemFunction("test", JoinFunction.class);
        //
        StringBuffer ddl = new StringBuffer();
        ddl.append("CREATE TABLE user_base (")
                .append("ub_id          BIGINT,")
                .append("ub_login_name  STRING,")
                .append("ub_login_pwd   STRING,")
                .append("ub_begin_time  TIMESTAMP,")
                .append("PRIMARY KEY (ub_id) NOT ENFORCED")
                .append(") WITH (")
                .append("'connector' = 'mysql-cdc', ")
                .append("'hostname' = '127.0.0.1', ")
                .append("'username' = 'root', ")
                .append("'password' = 'root', ")
                .append("'database-name' = 'ds_canal', ")
                .append("'table-name'= 'user_base', ")
                .append("'port' = '3306', ")
                .append("'server-id' = '1', ")
                .append("'server-time-zone' = 'Asia/Shanghai', ")
                .append("'debezium.database.whitelist' = 'ds_canal', ")
                .append("'debezium.table.whitelist' = 'ds_canal.user_base', ")
                .append("'debezium.include.schema.changes' = 'false', ")
                .append("'debezium.snapshot.locking.mode' = 'none'")
                .append(")");
        TableResult tblResult = tableEnv.executeSql(ddl.toString());
        //========== 【查询表】 ==========
        StringBuffer sql = new StringBuffer();
        sql.append("select ub_id, ")
                .append("ub_login_name, ")
                .append("w, ")
                .append("len ")
                .append("from user_base ub ")
                .append("join LATERAL TABLE (test(ub_login_name)) ON TRUE ");
        Table table = tableEnv.sqlQuery(sql.toString());
        //
        tableEnv.toRetractStream(table, Row.class).print(">>>>>>>>>>>>>>>>>>>>");

        //
        env.execute("cdc-debezium-query-job");
    }
}
